package com.myz.rabbitmq.confirm.asyn;

import com.myz.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

/**
 * 异步消息模式
 * addConfirmListener(ConfirmListener listener)
 *
 * @author maoyz on 18-4-11.
 */
public class AsynSend {

    private final static String QUEUE_NAME = "confirm_asyn_work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        // 有序集合SortedSet,unconfirm的消息序号集合
        final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
        //　声明一个持久化的 队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 开启confirm模式
        channel.confirmSelect();

        // 添加ConfirmListener
        channel.addConfirmListener(new ConfirmListener() {

            // 处理返回确认成功
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("消息发送成功,deliveryTag:" + deliveryTag + ",multiple:" + multiple + "");
                if (multiple) {
                    // 每publish一条数据，集合中元素加1，每回调一次handleAck方法，
                    // unconfirm集合删掉相应的一条(multiple=false) 或多条(multiple=true)记录
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }

            }

            // 处理返回确认失败
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("失败,deliveryTag:" + deliveryTag + "multiple:" + multiple + "");
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }
        });

        for (int i = 0; i < 100; i++) {
            String msg = "AsynSend ..." + i;
            // 获取
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            System.out.println(nextPublishSeqNo);
            // 每publish一条数据，集合中元素加1，
            confirmSet.add(nextPublishSeqNo);
        }
        channel.close();
        connection.close();
    }

}
